package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.mushroom.Slave;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * @ClassName: SessSourBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:10 
 *
 */
public class SessSourBolt extends AbstractBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = -5475897483936259084L;

	HashMap<String,Long> abnSessSourDMap = new HashMap<String,Long>();
	HashMap<String,Long> streamPacketMap = new HashMap<String,Long>();
	HashMap<String,Long> streamOctetsMap = new HashMap<String,Long>();

	HashSet<String> IPSet = new HashSet<String>();
	Slave s = null;

	@Override
	public void cleanup() {

	}

	@Override
	public void execute(Tuple input) {
		try {

			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				collector.ack(input);
			}
		} catch (Exception e) {
			Logger.getRootLogger().error(e);
			throw new FailedException("BaseStatSumBolt Exception",e);
		}	

	}

	private void countInLocal(Tuple tuple) {

		String dip = tuple.getStringByField("DIP");

		String sip = tuple.getStringByField("SIP");

		String sourceArea = tuple.getStringByField("AREANAME");

		String sourceCountry = tuple.getStringByField("SOURCECOUNTRY");
		
		long nPV = tuple.getLongByField("PV");

		long streamOctets = tuple.getLongByField("STREAMOCTETS");

		long streamPacket = tuple.getLongByField("STREAMPACKET");

		String abnType = tuple.getStringByField("ABNTYPE");


		Date date = new Date();
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		String dateStr_d = sdf_d.format(date);



		//###天
		String rowkeyDay = String.format("%s|%s|%s|%s|%s|%s", dip, dateStr_d , sip,sourceArea,sourceCountry,abnType);
		if(abnSessSourDMap.containsKey(rowkeyDay))
		{
			long dipPV = 0;
			long dipOctects = 0L;
			long dipPacket = 0L;
			dipPV = abnSessSourDMap.get(rowkeyDay) + nPV;
			dipOctects = streamOctetsMap.get(rowkeyDay) + streamOctets;
			dipPacket = streamPacketMap.get(rowkeyDay) + streamPacket;
			abnSessSourDMap.put(rowkeyDay, dipPV);
			streamOctetsMap.put(rowkeyDay, dipOctects);
			streamPacketMap.put(rowkeyDay, dipPacket);
		}
		else
		{
			abnSessSourDMap.put(rowkeyDay, nPV);
			streamOctetsMap.put(rowkeyDay, streamOctets);
			streamPacketMap.put(rowkeyDay, streamPacket);
		}

	}

	private void emitCountingData() {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);

		if(s.GetIPFilterForLog() != null && s.GetIPFilterForLog().size() > 0) {
			IPSet = s.GetIPFilterForLog();
		}

		for(String key : abnSessSourDMap.keySet()) {
			String[] split = key.split("\\|",-1);
			String dip = split[0];
			String dateStr_d = split[1];
			String sip = split[2];
			String sourceArea = split[3];
			String sourceCountry = split[4];
			String type = split[5];
			long abnSourDPV = abnSessSourDMap.get(key);
			long streamONum = streamOctetsMap.get(key);
			long streamPNum = streamPacketMap.get(key);

			if(IPSet.size()==0 || IPSet.contains(dip)) {

				String rowkeyDay = String.format("%s_%s_%s", dip, dateStr_d, sip);
				String tbNameDay = ApolloProperties.SDS_ABN_SESS_SOUR_STAT_D;
				hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "DESTIP", dip);
				hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "REPORTTIME", dateStr_d);
				hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "SOURCEIP", sip);
				hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "SOURCEAREA", sourceArea);
				hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "SOURCECOUNTRY", sourceCountry);
				hbaseInstance.incerment(tbNameDay, rowkeyDay, "cf:PV", streamPNum);

//				if(!"".equals(type) && streamONum > 0 && streamPNum > 0) {
					//端口单独记次数处理
					/**hbaseInstance.Add(tbNameDay, rowkeyDay, "cf", "ABNTYPE", type);
					System.out.println("################SessSourBolt streamOnum:" + streamONum + ",streamPNum:" + streamPNum + ";");
					 **/
					hbaseInstance.incerment(tbNameDay, rowkeyDay, "cf:STREAMOCTETS", streamONum);
					hbaseInstance.incerment(tbNameDay, rowkeyDay, "cf:STREAMPACKET", streamPNum);
//				}


			}
		}

		abnSessSourDMap.clear();
		streamOctetsMap.clear();
		streamPacketMap.clear();

	}



	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,600);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		s = new Slave(zooserver);
		IPSet = s.GetIPFilterForLog();
	}


}
